package com.danbay.framework.kafka.config;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.*;
import org.springframework.beans.factory.config.*;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.expression.StandardBeanExpressionResolver;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.format.support.DefaultFormattingConversionService;
import org.springframework.kafka.annotation.*;
import org.springframework.kafka.config.*;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.messaging.converter.GenericMessageConverter;
import org.springframework.messaging.handler.annotation.support.*;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.stereotype.Controller;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;

import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

/**
 * @author SJG
 *         2017/11/15.
 */
@ConfigurationProperties
public class KafkaListenerBeanPostProcessor<K, V> implements BeanPostProcessor, Ordered, BeanFactoryAware, SmartInitializingSingleton, EnvironmentAware {
    private final Set<Class<?>> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap(64));
    private final Log logger = LogFactory.getLog(this.getClass());
    private KafkaListenerEndpointRegistry endpointRegistry;
    private String containerFactoryBeanName = "kafkaListenerContainerFactory";
    private BeanFactory beanFactory;
    private final KafkaListenerBeanPostProcessor.KafkaHandlerMethodFactoryAdapter messageHandlerMethodFactory = new KafkaListenerBeanPostProcessor.KafkaHandlerMethodFactoryAdapter();
    private final KafkaListenerEndpointRegistrar registrar = new KafkaListenerEndpointRegistrar();
    private final AtomicInteger counter = new AtomicInteger();
    private BeanExpressionResolver resolver = new StandardBeanExpressionResolver();
    private BeanExpressionContext expressionContext;

    /**
     * 监听的主题列表
     */
    private String[] defaultListenTopics;
    private String consumerGroupId;

    public KafkaListenerBeanPostProcessor() {
    }

    @Override
    public int getOrder() {
        return 2147483647;
    }

    public void setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry) {
        this.endpointRegistry = endpointRegistry;
    }

    public void setContainerFactoryBeanName(String containerFactoryBeanName) {
        this.containerFactoryBeanName = containerFactoryBeanName;
    }

    public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory messageHandlerMethodFactory) {
        this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(messageHandlerMethodFactory);
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
        if (beanFactory instanceof ConfigurableListableBeanFactory) {
            this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver();
            this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory, null);
        }

    }

    @Override
    public void afterSingletonsInstantiated() {
        this.registrar.setBeanFactory(this.beanFactory);
        if (this.beanFactory instanceof ListableBeanFactory) {
            Map<String, KafkaListenerConfigurer> instances = ((ListableBeanFactory) this.beanFactory).getBeansOfType(KafkaListenerConfigurer.class);
            Iterator var2 = instances.values().iterator();

            while (var2.hasNext()) {
                KafkaListenerConfigurer configurer = (KafkaListenerConfigurer) var2.next();
                configurer.configureKafkaListeners(this.registrar);
            }
        }

        if (this.registrar.getEndpointRegistry() == null) {
            if (this.endpointRegistry == null) {
                Assert.state(this.beanFactory != null, "BeanFactory must be set to find endpoint registry by bean name");
                this.endpointRegistry = this.beanFactory.getBean("org.springframework.kafka.config.internalKafkaListenerEndpointRegistry", KafkaListenerEndpointRegistry.class);
            }

            this.registrar.setEndpointRegistry(this.endpointRegistry);
        }

        if (this.containerFactoryBeanName != null) {
            this.registrar.setContainerFactoryBeanName(this.containerFactoryBeanName);
        }

        MessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory();
        if (handlerMethodFactory != null) {
            this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(handlerMethodFactory);
        }

        this.registrar.afterPropertiesSet();
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        //找到所有controller
        if (
                bean.getClass().getAnnotation(Controller.class) != null ||
                        bean.getClass().getAnnotation(RequestMapping.class) != null
                ) {

        }

        if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
            Class<?> targetClass = AopUtils.getTargetClass(bean);
            Collection<KafkaListener> classLevelListeners = this.findListenerAnnotations(targetClass);
            boolean hasClassLevelListeners = classLevelListeners.size() > 0;
            List<Method> multiMethods = new ArrayList();
            Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass, new MethodIntrospector.MetadataLookup<Set<KafkaListener>>() {
                @Override
                public Set<KafkaListener> inspect(Method method) {
                    Set<KafkaListener> listenerMethods = KafkaListenerBeanPostProcessor.this.findListenerAnnotations(method);
                    return !listenerMethods.isEmpty() ? listenerMethods : null;
                }
            });
            if (hasClassLevelListeners) {
                Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass, new ReflectionUtils.MethodFilter() {
                    @Override
                    public boolean matches(Method method) {
                        return AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null;
                    }
                });
                multiMethods.addAll(methodsWithHandler);
            }

            if (annotatedMethods.isEmpty()) {
                this.nonAnnotatedClasses.add(bean.getClass());
                if (this.logger.isTraceEnabled()) {
                    this.logger.trace("No @KafkaListener annotations found on bean type: " + bean.getClass());
                }
            } else {
                Iterator var13 = annotatedMethods.entrySet().iterator();

                while (true) {
                    if (!var13.hasNext()) {
                        if (this.logger.isDebugEnabled()) {
                            this.logger.debug(annotatedMethods.size() + " @KafkaListener methods processed on bean '" + beanName + "': " + annotatedMethods);
                        }
                        break;
                    }

                    Map.Entry<Method, Set<KafkaListener>> entry = (Map.Entry) var13.next();
                    Method method = entry.getKey();

                    for (Object o : entry.getValue()) {
                        KafkaListener listener = (KafkaListener) o;
                        this.processKafkaListener(listener, method, bean, beanName);
                    }
                }
            }

            if (hasClassLevelListeners) {
                this.processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);
            }
        }

        return bean;
    }

    private Collection<KafkaListener> findListenerAnnotations(Class<?> clazz) {
        HashSet listeners = new HashSet();
        KafkaListener ann = AnnotationUtils.findAnnotation(clazz, KafkaListener.class);
        if (ann != null) {
            listeners.add(ann);
        }

        KafkaListeners ands = AnnotationUtils.findAnnotation(clazz, KafkaListeners.class);
        if (ands != null) {
            listeners.addAll(Arrays.asList(ands.value()));
        }

        return listeners;
    }

    private Set<KafkaListener> findListenerAnnotations(Method method) {
        HashSet listeners = new HashSet();
        KafkaListener ann = AnnotationUtils.findAnnotation(method, KafkaListener.class);
        if (ann != null) {
            listeners.add(ann);
        }

        KafkaListeners anns = AnnotationUtils.findAnnotation(method, KafkaListeners.class);
        if (anns != null) {
            listeners.addAll(Arrays.asList(anns.value()));
        }

        return listeners;
    }

    private void processMultiMethodListeners(Collection<KafkaListener> classLevelListeners, List<Method> multiMethods, Object bean, String beanName) {
        ArrayList checkedMethods = new ArrayList();
        Iterator var6 = multiMethods.iterator();

        while (var6.hasNext()) {
            Method method = (Method) var6.next();
            checkedMethods.add(this.checkProxy(method, bean));
        }

        var6 = classLevelListeners.iterator();

        while (var6.hasNext()) {
            KafkaListener classLevelListener = (KafkaListener) var6.next();
            MultiMethodKafkaListenerEndpoint endpoint = new MultiMethodKafkaListenerEndpoint(checkedMethods, bean);
            endpoint.setBeanFactory(this.beanFactory);
            this.processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);
        }

    }

    private void processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName) {
        Method methodToUse = this.checkProxy(method, bean);
        MethodKafkaListenerEndpoint endpoint = new MethodKafkaListenerEndpoint();
        endpoint.setMethod(methodToUse);
        endpoint.setBeanFactory(this.beanFactory);
        this.processListener(endpoint, kafkaListener, bean, methodToUse, beanName);
    }

    private Method checkProxy(Method methodArg, Object bean) {
        Method method = methodArg;
        if (AopUtils.isJdkDynamicProxy(bean)) {
            try {
                method = bean.getClass().getMethod(method.getName(), method.getParameterTypes());
                Class<?>[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces();
                int var6 = proxiedInterfaces.length;
                int var7 = 0;

                while (var7 < var6) {
                    Class iface = proxiedInterfaces[var7];
                    try {
                        method = iface.getMethod(method.getName(), method.getParameterTypes());
                        break;
                    } catch (NoSuchMethodException var10) {
                        ++var7;
                    }
                }
            } catch (SecurityException var11) {
                ReflectionUtils.handleReflectionException(var11);
            } catch (NoSuchMethodException var12) {
                throw new IllegalStateException(String.format("@KafkaListener method '%s' found on bean target class '%s', but not found in any interface(s) for bean JDK proxy. Either pull the method up to an interface or switch to subclass (CGLIB) proxies by setting proxy-target-class/proxyTargetClass attribute to 'true'", methodArg.getName(), methodArg.getDeclaringClass().getSimpleName()), var12);
            }
        }

        return method;
    }

    private void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener, Object bean, Object adminTarget, String beanName) {
        endpoint.setBean(bean);
        endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
        endpoint.setId(this.getEndpointId(kafkaListener));
        endpoint.setTopicPartitions(this.resolveTopicPartitions(kafkaListener));
        endpoint.setTopics(this.resolveTopics(kafkaListener));
        endpoint.setTopicPattern(this.resolvePattern(kafkaListener));
        resolveDefaultTopic(endpoint);

        String group = kafkaListener.group();
        if (StringUtils.hasText(group)) {
            Object resolvedGroup = this.resolveExpression(group);
            if (resolvedGroup instanceof String) {
                endpoint.setGroup((String) resolvedGroup);
            }
        }

        KafkaListenerContainerFactory<?> factory = null;
        String containerFactoryBeanName = this.resolve(kafkaListener.containerFactory());
        if (StringUtils.hasText(containerFactoryBeanName)) {
            Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");

            try {
                factory = (KafkaListenerContainerFactory) this.beanFactory.getBean(containerFactoryBeanName, KafkaListenerContainerFactory.class);
            } catch (NoSuchBeanDefinitionException var10) {
                throw new BeanInitializationException("Could not register Kafka listener endpoint on [" + adminTarget + "] for bean " + beanName + ", no " + KafkaListenerContainerFactory.class.getSimpleName() + " with id '" + containerFactoryBeanName + "' was found in the application context", var10);
            }
        }

        this.registrar.registerEndpoint(endpoint, factory);
    }

    private void resolveDefaultTopic(MethodKafkaListenerEndpoint<?, ?> endpoint) {
        //没有设置topic也没有设置topic匹配规则，则使用配置文件配置的主题
        if (endpoint.getTopics().isEmpty() && endpoint.getTopicPattern() == null) {
            endpoint.setTopics(defaultListenTopics);
            endpoint.setGroup(consumerGroupId);
        }
    }

    private String getEndpointId(KafkaListener kafkaListener) {
        return StringUtils.hasText(kafkaListener.id()) ? this.resolve(kafkaListener.id()) : "org.springframework.kafka.KafkaListenerEndpointContainer#" + this.counter.getAndIncrement();
    }

    private TopicPartitionInitialOffset[] resolveTopicPartitions(KafkaListener kafkaListener) {
        TopicPartition[] topicPartitions = kafkaListener.topicPartitions();
        List<TopicPartitionInitialOffset> result = new ArrayList();
        if (topicPartitions.length > 0) {

            for (TopicPartition topicPartition : topicPartitions) {
                result.addAll(this.resolveTopicPartitionsList(topicPartition));
            }
        }

        return result.toArray(new TopicPartitionInitialOffset[result.size()]);
    }

    private String[] resolveTopics(KafkaListener kafkaListener) {
        String[] topics = kafkaListener.topics();
        ArrayList result = new ArrayList();
        if (topics.length > 0) {
            for (String topic1 : topics) {
                Object topic = this.resolveExpression(topic1);
                this.resolveAsString(topic, result);
            }
        }

        return (String[]) result.toArray(new String[result.size()]);
    }

    private Pattern resolvePattern(KafkaListener kafkaListener) {
        Pattern pattern = null;
        String text = kafkaListener.topicPattern();
        if (StringUtils.hasText(text)) {
            Object resolved = this.resolveExpression(text);
            if (resolved instanceof Pattern) {
                pattern = (Pattern) resolved;
            } else {
                if (!(resolved instanceof String)) {
                    throw new IllegalStateException("topicPattern must resolve to a Pattern or String, not " + resolved.getClass());
                }

                pattern = Pattern.compile((String) resolved);
            }
        }

        return pattern;
    }

    private ArrayList resolveTopicPartitionsList(TopicPartition topicPartition) {
        Object topic = this.resolveExpression(topicPartition.topic());
        Assert.state(topic instanceof String, "topic in @TopicPartition must resolve to a String, not " + topic.getClass());
        Assert.state(StringUtils.hasText((String) topic), "topic in @TopicPartition must not be empty");
        String[] partitions = topicPartition.partitions();
        PartitionOffset[] partitionOffsets = topicPartition.partitionOffsets();
        Assert.state(partitions.length > 0 || partitionOffsets.length > 0, "At least one 'partition' or 'partitionOffset' required in @TopicPartition for topic '" + topic + "'");
        ArrayList result = new ArrayList();

        for (String partition1 : partitions) {
            this.resolvePartitionAsInteger((String) topic, this.resolveExpression(partition1), result);
        }

        int var7 = partitionOffsets.length;

        for (int var8 = 0; var8 < var7; ++var8) {
            PartitionOffset partitionOffset = partitionOffsets[var8];
            Object partitionValue = this.resolveExpression(partitionOffset.partition());
            Integer partition;
            if (partitionValue instanceof String) {
                Assert.state(StringUtils.hasText((String) partitionValue), "partition in @PartitionOffset for topic '" + topic + "' cannot be empty");
                partition = Integer.valueOf((String) partitionValue);
            } else {
                if (!(partitionValue instanceof Integer)) {
                    throw new IllegalArgumentException(String.format("@PartitionOffset for topic '%s' can't resolve '%s' as an Integer or String, resolved to '%s'", new Object[]{topic, partitionOffset.partition(), partitionValue.getClass()}));
                }

                partition = (Integer) partitionValue;
            }

            Object initialOffsetValue = this.resolveExpression(partitionOffset.initialOffset());
            Long initialOffset;
            if (initialOffsetValue instanceof String) {
                Assert.state(StringUtils.hasText((String) initialOffsetValue), "'initialOffset' in @PartitionOffset for topic '" + topic + "' cannot be empty");
                initialOffset = Long.valueOf((String) initialOffsetValue);
            } else {
                if (!(initialOffsetValue instanceof Long)) {
                    throw new IllegalArgumentException(String.format("@PartitionOffset for topic '%s' can't resolve '%s' as a Long or String, resolved to '%s'", new Object[]{topic, partitionOffset.initialOffset(), initialOffsetValue.getClass()}));
                }

                initialOffset = (Long) initialOffsetValue;
            }

            Object relativeToCurrentValue = this.resolveExpression(partitionOffset.relativeToCurrent());
            Boolean relativeToCurrent;
            if (relativeToCurrentValue instanceof String) {
                relativeToCurrent = Boolean.valueOf((String) relativeToCurrentValue);
            } else {
                if (!(relativeToCurrentValue instanceof Boolean)) {
                    throw new IllegalArgumentException(String.format("@PartitionOffset for topic '%s' can't resolve '%s' as a Boolean or String, resolved to '%s'", new Object[]{topic, partitionOffset.relativeToCurrent(), relativeToCurrentValue.getClass()}));
                }

                relativeToCurrent = (Boolean) relativeToCurrentValue;
            }

            TopicPartitionInitialOffset topicPartitionOffset = new TopicPartitionInitialOffset((String) topic, partition.intValue(), initialOffset, relativeToCurrent.booleanValue());
            if (result.contains(topicPartitionOffset)) {
                throw new IllegalArgumentException(String.format("@TopicPartition can't have the same partition configuration twice: [%s]", new Object[]{topicPartitionOffset}));
            }

            result.add(topicPartitionOffset);
        }

        return result;
    }

    private void resolveAsString(Object resolvedValue, List<String> result) {
        if (resolvedValue instanceof String[]) {
            String[] var3 = (String[]) resolvedValue;

            for (String object : var3) {
                this.resolveAsString(object, result);
            }
        }

        if (resolvedValue instanceof String) {
            result.add((String) resolvedValue);
        } else {
            if (!(resolvedValue instanceof Iterable)) {
                throw new IllegalArgumentException(String.format("@KafKaListener can't resolve '%s' as a String", new Object[]{resolvedValue}));
            }

            for (Object object : ((Iterable) resolvedValue)) {
                this.resolveAsString(object, result);
            }
        }

    }

    private void resolvePartitionAsInteger(String topic, Object resolvedValue, List<TopicPartitionInitialOffset> result) {
        int var5;
        int var6;
        if (resolvedValue instanceof String[]) {
            String[] var4 = (String[]) resolvedValue;
            var5 = var4.length;

            for (var6 = 0; var6 < var5; ++var6) {
                Object object = var4[var6];
                this.resolvePartitionAsInteger(topic, object, result);
            }
        } else if (resolvedValue instanceof String) {
            Assert.state(StringUtils.hasText((String) resolvedValue), "partition in @TopicPartition for topic '" + topic + "' cannot be empty");
            result.add(new TopicPartitionInitialOffset(topic, Integer.valueOf((String) resolvedValue)));
        } else if (resolvedValue instanceof Integer[]) {
            Integer[] var8 = (Integer[]) resolvedValue;
            var5 = var8.length;

            for (var6 = 0; var6 < var5; ++var6) {
                Integer partition = var8[var6];
                result.add(new TopicPartitionInitialOffset(topic, partition));
            }
        } else if (resolvedValue instanceof Integer) {
            result.add(new TopicPartitionInitialOffset(topic, (Integer) resolvedValue));
        } else {
            if (!(resolvedValue instanceof Iterable)) {
                throw new IllegalArgumentException(String.format("@KafKaListener for topic '%s' can't resolve '%s' as an Integer or String", new Object[]{topic, resolvedValue}));
            }

            for (Object object : ((Iterable) resolvedValue)) {
                this.resolvePartitionAsInteger(topic, object, result);
            }
        }

    }

    private Object resolveExpression(String value) {
        String resolvedValue = this.resolve(value);
        return resolvedValue.startsWith("#{") && value.endsWith("}") ? this.resolver.evaluate(resolvedValue, this.expressionContext) : resolvedValue;
    }

    private String resolve(String value) {
        return this.beanFactory != null && this.beanFactory instanceof ConfigurableBeanFactory ? ((ConfigurableBeanFactory) this.beanFactory).resolveEmbeddedValue(value) : value;
    }

    @Override
    public void setEnvironment(Environment environment) {
        //取得要订阅的命令列表
        Set<String> listenTopics = new HashSet<>();
        String subscribeConfig = environment.getProperty("spring.kafka.consumer.subscribe-topics");
        if (subscribeConfig != null) {
            for (String key : subscribeConfig.split(",")) {
                //从配置中取得要订阅的主题列表
                String topics = environment.getProperty(key);
                if(topics!=null){
                    listenTopics.addAll(Arrays.asList(topics.split(",")));
                }
            }
        }
        defaultListenTopics = listenTopics.toArray(new String[]{});
        consumerGroupId = environment.getProperty("spring.kafka.consumer.group-id");
    }

    private class KafkaHandlerMethodFactoryAdapter implements MessageHandlerMethodFactory {
        private MessageHandlerMethodFactory messageHandlerMethodFactory;

        private KafkaHandlerMethodFactoryAdapter() {
        }

        void setMessageHandlerMethodFactory(MessageHandlerMethodFactory kafkaHandlerMethodFactory1) {
            this.messageHandlerMethodFactory = kafkaHandlerMethodFactory1;
        }

        @Override
        public InvocableHandlerMethod createInvocableHandlerMethod(Object bean, Method method) {
            return this.getMessageHandlerMethodFactory().createInvocableHandlerMethod(bean, method);
        }

        private MessageHandlerMethodFactory getMessageHandlerMethodFactory() {
            if (this.messageHandlerMethodFactory == null) {
                this.messageHandlerMethodFactory = this.createDefaultMessageHandlerMethodFactory();
            }

            return this.messageHandlerMethodFactory;
        }

        private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
            DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();
            defaultFactory.setBeanFactory(KafkaListenerBeanPostProcessor.this.beanFactory);
            ConfigurableBeanFactory cbf = KafkaListenerBeanPostProcessor.this.beanFactory instanceof ConfigurableBeanFactory ? (ConfigurableBeanFactory) KafkaListenerBeanPostProcessor.this.beanFactory : null;
            DefaultFormattingConversionService conversionService = new DefaultFormattingConversionService();
            defaultFactory.setConversionService(conversionService);
            List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList();
            argumentResolvers.add(new HeaderMethodArgumentResolver(conversionService, cbf));
            argumentResolvers.add(new HeadersMethodArgumentResolver());
            GenericMessageConverter messageConverter = new GenericMessageConverter(conversionService);
            argumentResolvers.add(new MessageMethodArgumentResolver(messageConverter));
            argumentResolvers.add(new PayloadArgumentResolver(messageConverter) {
                @Override
                protected boolean isEmptyPayload(Object payload) {
                    return payload == null || payload instanceof KafkaNull;
                }
            });
            defaultFactory.setArgumentResolvers(argumentResolvers);
            defaultFactory.afterPropertiesSet();
            return defaultFactory;
        }
    }
}